package com.dingwen.treasure.mybatisplus.service.impl;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.dingwen.treasure.base.enums.MsgState;
import com.dingwen.treasure.base.util.ConvertUtil;
import com.dingwen.treasure.model.entity.mybatisplus.TreasureMessage;
import com.dingwen.treasure.mybatisplus.bo.NeedDeliverMessageBO;
import com.dingwen.treasure.mybatisplus.mapper.TreasureMessageMapper;
import com.dingwen.treasure.mybatisplus.service.TreasureMessageService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author dingwen
 * @description 针对表【treasure_message(消息表)】的数据库操作Service实现
 * @createDate 2022-08-09 13:18:00
 */
@Service
@Slf4j
public class TreasureMessageServiceImpl extends ServiceImpl<TreasureMessageMapper, TreasureMessage> implements TreasureMessageService {


    /**
     * messageMapper
     */
    @Resource
    private  TreasureMessageMapper messageMapper;


    /**
     * rabbitTemplate
     */
    @Resource
    private  RabbitTemplate rabbitTemplate;


    /**
     * 消息投递
     */
    @Override
    @Async(value = "asyncExecutor")
    public void deliverMessage() {
        ReentrantLock lock = new ReentrantLock();
        lock.lock();
        try {
            List<NeedDeliverMessageBO> messageBo = messageMapper.getNeedDeliverMessageInfo();
            messageBo.parallelStream()
                    .forEach(bo -> {
                        List<Integer> orders = ConvertUtil.strToList(Integer.class, bo.getOrders());
                        Integer order = orders.get(Convert.toInt(Math.ceil(bo.getTotalState() / 2.0)));
                        LambdaQueryWrapper<TreasureMessage> lambdaQuery = new LambdaQueryWrapper<>();
                        lambdaQuery.eq(TreasureMessage::getBusinessId, bo.getBusinessId())
                                .eq(TreasureMessage::getBusinessOrder, order);
                        TreasureMessage message = getOne(lambdaQuery);
                        if (Objects.isNull(message)) {
                            return;
                        }
                        // 消息发送
                        sendMessage(message);
                    });

        } finally {
            lock.unlock();
        }
    }


    /**
     * 更新消息状态
     *
     * @param mgState   mg状态
     * @param messageId 消息id
     */
    @Override
    public void updateMessageState(MsgState mgState, Long messageId) {
        // 此方式不会填充修改时间
        LambdaUpdateWrapper<TreasureMessage> updateWrapper = new LambdaUpdateWrapper<>();
        updateWrapper.eq(TreasureMessage::getMsgId, messageId)
                .set(TreasureMessage::getMsgState, mgState)
                .set(TreasureMessage::getUpdateTime, LocalDateTime.now());
        update(updateWrapper);
    }

    /**
     * 发送消息
     *
     * @param message 消息
     */
    private void sendMessage(TreasureMessage message) {
        String queue = message.getQueue();
        String router = message.getRouter();
        String exchange = message.getExchange();
        String messageId = Convert.toStr(message.getMsgId());
        String msgBody = message.getMsgBody();
        Assert.isTrue(StrUtil.isNotBlank(router), "路由信息不能为空");
        Assert.isTrue(StrUtil.isNotBlank(exchange), "交换机信息不能为空");
        Assert.isTrue(StrUtil.isNotBlank(msgBody), "消息体信息不能为空");
        Assert.isTrue(StrUtil.isNotBlank(queue), "队列信息不能为空");

        CorrelationData correlationData = new CorrelationData();
        correlationData.setId(messageId);
        // 交换机通过路由键投递到指定路由
        rabbitTemplate.convertAndSend(
                exchange,
                router,
                msgBody,
                correlationData);
        updateMessageState(MsgState.DELIVERED, Convert.toLong(messageId));
        log.info("消息投递成功，交换机：{},队列：{},路由：{},消息ID：{}", exchange, queue, router, messageId);

    }


}




